package zy.learn.demo.structuredstreaming.source

import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.streaming.{OutputMode, Trigger}
import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}

object FileSource2 {
  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().set("spark.sql.shuffle.partitions", "3")

    val spark = SparkSession.builder()
      .master("local[2]")
      .config(sparkConf)
      .appName("FileSource2")
      .getOrCreate()

    // 创建Schema
    val userSchema = StructType(StructField("name", StringType) :: StructField("age", IntegerType) :: StructField("sex", StringType) :: Nil)
    // 以流模式加载csv文件
    spark.readStream
      .format("csv")
      .schema(userSchema)
      .load("E:\\MyGit\\spark-project\\data\\ss")
      .createOrReplaceTempView("csvfile")

    spark.sql(
      """
        | select sex, count(1) as cnt
        |   from csvfile
        |  group by sex
        |""".stripMargin).createOrReplaceTempView("tmpView1")

    val resultDF = spark.sql(
      """
        | select sex as SEX, cnt as CNT from tmpView1
        |""".stripMargin)
    // 输出
    resultDF.writeStream
      .format("console")
      .outputMode(OutputMode.Update())
      .trigger(Trigger.ProcessingTime(1000))
      .start()
      .awaitTermination()
  }
}
